package com.jkgl.function;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.jkgl.entity.MesPersistence;
import com.jkgl.service.MesPersistenceService;
import com.jkgl.service.SendService;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
@Slf4j
public class ErrorConsumerService  implements StreamListener<String, ObjectRecord<String, String>> {
    private final SendService sendService;
    private final MesPersistenceService persistenceService;
    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Override
    public void onMessage(ObjectRecord<String, String> message) {
//        TODO 不会对已经消费的信息进行删除 --做数据库存储以及删除
//        String stream = message.getStream();
        String messageId = message.getId().toString();
//        String messageBody = message.getValue();
        log.info("message:{}",message);
//        System.out.println(stream);
//        System.out.println(messageId);
//        System.out.println(messageBody);
        QueryWrapper<MesPersistence> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MesPersistence::getMessageId,messageId);
        MesPersistence one = persistenceService.getOne(queryWrapper);
        boolean result=sendService.sendError(message);
        one.setIsSuccess(2);
        one.setIsSend(result ? 1:2);
        persistenceService.updateById(one);
        log.info("删除error中数据{}",message.getId());
        redisTemplate.opsForStream().delete("jkgl-error",message.getId().toString());
    }
}
